
根据提供的项目规范和源代码，为 `DsoftbusInputPlugin::OnStreamReceived` 函数生成单元测试代码：

```cpp
/*
 * Copyright (c) 2025 Huawei Device Co., Ltd.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "dsoftbus_input_plugin.h"
#include <gtest/gtest.h>
#include <gmock/gmock.h>
#include <memory>
#include "av_trans_constants.h"

using namespace testing;
using namespace testing::ext;

namespace OHOS {
namespace DistributedHardware {

class DsoftbusInputPluginTest : public testing::Test {
public:
    static void SetUpTestCase(void);
    static void TearDownTestCase(void);
    void SetUp();
    void TearDown();
    
    std::shared_ptr<DsoftbusInputPlugin> instance_;
};

void DsoftbusInputPluginTest::SetUpTestCase(void)
{
    // 所有测试开始前执行一次
}

void DsoftbusInputPluginTest::TearDownTestCase(void)
{
    // 所有测试结束后执行一次
}

void DsoftbusInputPluginTest::SetUp()
{
    instance_ = std::make_shared<DsoftbusInputPlugin>("test_plugin");
    ASSERT_NE(instance_, nullptr);
    instance_->Init();
    instance_->SetDumpFlagFalse();
    instance_->SetReDumpFlagFalse();
    GTEST_LOG_(INFO) << "SetUp";
}

void DsoftbusInputPluginTest::TearDown()
{
    if (instance_ != nullptr) {
        instance_->Reset();
    }
    instance_ = nullptr;
    GTEST_LOG_(INFO) << "TearDown";
}

/**
 * @tc.number: DsoftbusInputPlugin_OnStreamReceived_001
 * @tc.name: OnStreamReceived
 * @tc.desc: Verify behavior when ext is nullptr
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_OnStreamReceived_001, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_OnStreamReceived_001";
    
    StreamData data = { "test_data", 9 };
    instance_->OnStreamReceived(&data, nullptr);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_OnStreamReceived_001";
}

/**
 * @tc.number: DsoftbusInputPlugin_OnStreamReceived_002
 * @tc.name: OnStreamReceived
 * @tc.desc: Verify behavior when ext contains invalid JSON
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_OnStreamReceived_002, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_OnStreamReceived_002";
    
    StreamData data = { "test_data", 9 };
    std::string invalidJson = R"({"invalid": json})";
    StreamData ext = { invalidJson.c_str(), static_cast<int32_t>(invalidJson.length()) };
    
    instance_->OnStreamReceived(&data, &ext);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_OnStreamReceived_002";
}

/**
 * @tc.number: DsoftbusInputPlugin_OnStreamReceived_003
 * @tc.name: OnStreamReceived
 * @tc.desc: Verify behavior when JSON lacks metaType field
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_OnStreamReceived_003, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_OnStreamReceived_003";
    
    StreamData data = { "test_data", 9 };
    std::string jsonWithoutMeta = R"({"param": "{}"})";
    StreamData ext = { jsonWithoutMeta.c_str(), static_cast<int32_t>(jsonWithoutMeta.length()) };
    
    instance_->OnStreamReceived(&data, &ext);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_OnStreamReceived_003";
}

/**
 * @tc.number: DsoftbusInputPlugin_OnStreamReceived_004
 * @tc.name: OnStreamReceived
 * @tc.desc: Verify behavior when metaType is not a valid uint32
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_OnStreamReceived_004, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_OnStreamReceived_004";
    
    StreamData data = { "test_data", 9 };
    std::string jsonWithInvalidMeta = R"({"metaType": "not_a_number", "param": "{}"})";
    StreamData ext = { jsonWithInvalidMeta.c_str(), static_cast<int32_t>(jsonWithInvalidMeta.length()) };
    
    instance_->OnStreamReceived(&data, &ext);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_OnStreamReceived_004";
}

/**
 * @tc.number: DsoftbusInputPlugin_OnStreamReceived_005
 * @tc.name: OnStreamReceived
 * @tc.desc: Verify behavior when data is nullptr
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_OnStreamReceived_005, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_OnStreamReceived_005";
    
    std::string validJson = R"({"metaType": 1, "param": "{}"})";
    StreamData ext = { validJson.c_str(), static_cast<int32_t>(validJson.length()) };
    
    instance_->OnStreamReceived(nullptr, &ext);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_OnStreamReceived_005";
}

/**
 * @tc.number: DsoftbusInputPlugin_OnStreamReceived_006
 * @tc.name: OnStreamReceived
 * @tc.desc: Verify behavior with valid data and ext
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_OnStreamReceived_006, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_OnStreamReceived_006";
    
    std::string testData = "test_data_buffer";
    StreamData data = { testData.c_str(), static_cast<int32_t>(testData.length()) };
    std::string validJson = R"({"metaType": 1, "param": "{}"})";
    StreamData ext = { validJson.c_str(), static_cast<int32_t>(validJson.length()) };
    
    instance_->OnStreamReceived(&data, &ext);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_OnStreamReceived_006";
}

/**
 * @tc.number: DsoftbusInputPlugin_OnStreamReceived_007
 * @tc.name: OnStreamReceived
 * @tc.desc: Verify behavior with empty data buffer
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_OnStreamReceived_007, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_OnStreamReceived_007";
    
    StreamData data = { "", 0 };
    std::string validJson = R"({"metaType": 1, "param": "{}"})";
    StreamData ext = { validJson.c_str(), static_cast<int32_t>(validJson.length()) };
    
    instance_->OnStreamReceived(&data, &ext);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_OnStreamReceived_007";
}

} // namespace DistributedHardware

/**
 * @tc.number: DsoftbusInputPlugin_CreateBuffer_001
 * @tc.name: CreateBuffer - Null Data Pointer
 * @tc.desc: Verify behavior when data pointer is null
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_CreateBuffer_001, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_CreateBuffer_001";
    
    // Arrange: Create valid ext JSON
    std::string jsonStr = R"({"type":0, "param":"{\"pts_\":1000, \"frameNum_\":1}"})";
    testExt_.buf = jsonStr.c_str();
    testExt_.bufLen = jsonStr.length();

    // Act: Call OnStreamReceived with null data
    plugin_->OnStreamReceived(nullptr, &testExt_);

    // Assert: No buffer should be created
    EXPECT_EQ(capturedBuffer_, nullptr);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_CreateBuffer_001";
}

/**
 * @tc.number: DsoftbusInputPlugin_CreateBuffer_002
 * @tc.name: CreateBuffer - Missing Param Field
 * @tc.desc: Verify behavior when JSON lacks param field
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_CreateBuffer_002, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_CreateBuffer_002";
    
    // Arrange: Create JSON without param field
    std::string jsonStr = R"({"type":0, "other":"value"})";
    testExt_.buf = jsonStr.c_str();
    testExt_.bufLen = jsonStr.length();
    
    // Prepare valid data
    std::string dataStr = "test_data";
    testData_.buf = dataStr.c_str();
    testData_.bufLen = dataStr.length();

    // Act
    plugin_->OnStreamReceived(&testData_, &testExt_);

    // Assert
    EXPECT_EQ(capturedBuffer_, nullptr);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_CreateBuffer_002";
}

/**
 * @tc.number: DsoftbusInputPlugin_CreateBuffer_003
 * @tc.name: CreateBuffer - Invalid Param Type
 * @tc.desc: Verify behavior when param field is not a string
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_CreateBuffer_003, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_CreateBuffer_003";
    
    // Arrange: Create JSON with non-string param
    std::string jsonStr = R"({"type":0, "param":12345})";
    testExt_.buf = jsonStr.c_str();
    testExt_.bufLen = jsonStr.length();
    
    std::string dataStr = "test_data";
    testData_.buf = dataStr.c_str();
    testData_.bufLen = dataStr.length();

    // Act
    plugin_->OnStreamReceived(&testData_, &testExt_);

    // Assert
    EXPECT_EQ(capturedBuffer_, nullptr);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_CreateBuffer_003";
}

/**
 * @tc.number: DsoftbusInputPlugin_CreateBuffer_004
 * @tc.name: CreateBuffer - Invalid Meta JSON
 * @tc.desc: Verify behavior when meta JSON cannot be unmarshaled
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_CreateBuffer_004, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_CreateBuffer_004";
    
    // Arrange: Create JSON with invalid meta string
    std::string jsonStr = R"({"type":0, "param":"{invalid_json}"})";
    testExt_.buf = jsonStr.c_str();
    testExt_.bufLen = jsonStr.length();
    
    std::string dataStr = "test_data";
    testData_.buf = dataStr.c_str();
    testData_.bufLen = dataStr.length();

    // Act
    plugin_->OnStreamReceived(&testData_, &testExt_);

    // Assert
    EXPECT_EQ(capturedBuffer_, nullptr);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_CreateBuffer_004";
}

/**
 * @tc.number: DsoftbusInputPlugin_CreateBuffer_005
 * @tc.name: CreateBuffer - Valid Meta Without Extension
 * @tc.desc: Verify buffer creation with valid meta but no extension data
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_CreateBuffer_005, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_CreateBuffer_005";
    
    // Arrange: Create valid JSON without extension fields
    std::string jsonStr = R"({"type":0, "param":"{\"pts_\":1000, \"frameNum_\":42}"})";
    testExt_.buf = jsonStr.c_str();
    testExt_.bufLen = jsonStr.length();
    
    std::string dataStr = "video_frame_data";
    testData_.buf = dataStr.c_str();
    testData_.bufLen = dataStr.length();

    // Act
    plugin_->OnStreamReceived(&testData_, &testExt_);

    // Assert
    EXPECT_NE(capturedBuffer_, nullptr);
    EXPECT_EQ(capturedBuffer_->pts, 1000);
    
    // Verify frame number metadata
    ValueType frameNum;
    EXPECT_EQ(capturedBuffer_->GetBufferMeta()->GetMeta(Tag::USER_FRAME_NUMBER, frameNum), Status::OK);
    EXPECT_EQ(Plugin::AnyCast<size_t>(frameNum), 42);
    
    // Verify extension metadata is not present
    EXPECT_EQ(capturedBuffer_->GetBufferMeta()->GetMeta(Tag::MEDIA_START_TIME, frameNum), 
              Status::ERROR_NOT_EXISTED);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_CreateBuffer_005";
}

/**
 * @tc.number: DsoftbusInputPlugin_CreateBuffer_006
 * @tc.name: CreateBuffer - Valid Meta With Extension
 * @tc.desc: Verify buffer creation with valid meta including extension data
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_CreateBuffer_006, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_CreateBuffer_006";
    
    // Arrange: Create valid JSON with extension fields
    std::string jsonStr = R"({"type":0, "param":"{\"pts_\":2000, \"frameNum_\":100, " 
                          R"("extFrameNum_\":50, \"extPts_\":5000}"})";
    testExt_.buf = jsonStr.c_str();
    testExt_.bufLen = jsonStr.length();
    
    std::string dataStr = "h264_frame_data";
    testData_.buf = dataStr.c_str();
    testData_.bufLen = dataStr.length();

    // Act
    plugin_->OnStreamReceived(&testData_, &testExt_);

    // Assert
    EXPECT_NE(capturedBuffer_, nullptr);
    EXPECT_EQ(capturedBuffer_->pts, 2000);
    
    // Verify frame number metadata
    ValueType frameNum;
    EXPECT_EQ(capturedBuffer_->GetBufferMeta()->GetMeta(Tag::USER_FRAME_NUMBER, frameNum), Status::OK);
    EXPECT_EQ(Plugin::AnyCast<size_t>(frameNum), 100);
    
    // Verify extension metadata is present
    EXPECT_EQ(capturedBuffer_->GetBufferMeta()->GetMeta(Tag::MEDIA_START_TIME, frameNum), Status::OK);
    EXPECT_EQ(Plugin::AnyCast<int64_t>(frameNum), 5000);
    
    EXPECT_EQ(capturedBuffer_->GetBufferMeta()->GetMeta(Tag::AUDIO_SAMPLE_PER_FRAME, frameNum), Status::OK);
    EXPECT_EQ(Plugin::AnyCast<uint32_t>(frameNum), 50);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_CreateBuffer_006";
}

/**
 * @tc.number: DsoftbusInputPlugin_CreateBuffer_007
 * @tc.name: CreateBuffer - Data Write Failure
 * @tc.desc: Verify behavior when data write to buffer fails
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_CreateBuffer_007, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_CreateBuffer_007";
    
    // Arrange: Create valid JSON
    std::string jsonStr = R"({"type":0, "param":"{\"pts_\":3000, \"frameNum_\":1}"})";
    testExt_.buf = jsonStr.c_str();
    testExt_.bufLen = jsonStr.length();
    
    // Create large data that may fail to write
    const size_t largeSize = 10 * 1024 * 1024; // 10MB
    auto largeData = std::make_unique<char[]>(largeSize);
    std::memset(largeData.get(), 'A', largeSize);
    testData_.buf = largeData.get();
    testData_.bufLen = largeSize;

    // Act
    plugin_->OnStreamReceived(&testData_, &testExt_);

    // Assert: Buffer should still be created even if write fails
    EXPECT_NE(capturedBuffer_, nullptr);
    EXPECT_EQ(capturedBuffer_->pts, 3000);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_CreateBuffer_007";
}

/**
 * @tc.number: DsoftbusInputPlugin_CreateBuffer_008
 * @tc.name: CreateBuffer - Zero Length Data
 * @tc.desc: Verify behavior with zero-length data
 */
HWTEST_F(DsoftbusInputPluginTest, DsoftbusInputPlugin_CreateBuffer_008, TestSize.Level1)
{
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-begin DsoftbusInputPlugin_CreateBuffer_008";
    
    // Arrange: Create valid JSON
    std::string jsonStr = R"({"type":0, "param":"{\"pts_\":4000, \"frameNum_\":1}"})";
    testExt_.buf = jsonStr.c_str();
    testExt_.bufLen = jsonStr.length();
    
    // Create zero-length data
    testData_.buf = "";
    testData_.bufLen = 0;

    // Act
    plugin_->OnStreamReceived(&testData_, &testExt_);

    // Assert: Buffer should be created with zero length
    EXPECT_NE(capturedBuffer_, nullptr);
    EXPECT_EQ(capturedBuffer_->pts, 4000);
    EXPECT_EQ(capturedBuffer_->GetMemory()->GetSize(), 0);
    
    GTEST_LOG_(INFO) << "DsoftbusInputPluginTest-end DsoftbusInputPlugin_CreateBuffer_008";
}

} // namespace OHOS
```

### 测试用例说明：

1. **OnStreamReceived_001**: 测试 `ext` 为 nullptr 的情况，验证函数安全返回
2. **OnStreamReceived_002**: 测试 ext 包含无效 JSON 的情况
3. **OnStreamReceived_003**: 测试 JSON 中缺少 metaType 字段的情况
4. **OnStreamReceived_004**: 测试 metaType 不是有效 uint32 的情况
5. **OnStreamReceived_005**: 测试 data 为 nullptr 的情况
6. **OnStreamReceived_006**: 测试正常有效的 data 和 ext 参数
7. **OnStreamReceived_007**: 测试空数据缓冲区的情况

### 测试覆盖重点：

1. **空指针检查**: 覆盖 `ext` 和 `data` 为 nullptr 的边界情况
2. **JSON 解析**: 验证无效 JSON 和字段缺失的处理
3. **类型验证**: 测试 metaType 类型不匹配的场景
4. **边界值**: 包含空缓冲区的测试
5. **正常流程**: 验证完整的数据处理流程

### 配套 BUILD.gn 需要添加：
```gn
ohos_unittest("dsoftbus_input_plugin_unittest") {
  module_out_path = "av_trans_engine/plugin/test"
  
  sources = [
    "dsoftbus_input_plugin_test.cpp",
  ]
  
  include_dirs = [
    "//foundation/distributedhardware/distributed_hardware_fwk/av_transport/av_trans_engine/plugin/plugins/av_trans_input/dsoftbus_input",
    "//foundation/distributedhardware/distributed_hardware_fwk/av_transport/av_trans_engine/plugin/interface",
  ]
  
  deps = [
    "//foundation/distributedhardware/distributed_hardware_fwk/av_transport/av_trans_engine/plugin/plugins/av_trans_input/dsoftbus_input:plugin_AVTransDsoftbusInput",
  ]
  
  external_deps = [
    "cJSON:cjson",
    "hilog:libhilog",
    "gtest:gtest_main",
  ]
  
  defines = [
    "HI_LOG_ENABLE",
    "DH_LOG_TAG=\"dsoftbus_input_plugin_test\"",
    "LOG_DOMAIN=0xD004101",
  ]
}
```